// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#pragma once

#include <algorithm>
#include <array>
#include <cstring>
#include <optional>
#include <string_view>
#include <type_traits>
#include <typeindex>
#include <typeinfo>
#include <utility>

#include <pollux/expression/complex_view_types.h>
#include <pollux/expression/decoded_args.h>
#include <pollux/expression/kind_to_simple_type.h>
#include <pollux/expression/udf_type_resolver.h>
#include <pollux/expression/variadic_view.h>
#include <pollux/common/strings/string_view.h>
#include <pollux/type/type.h>
#include <pollux/vector/base_vector.h>
#include <pollux/vector/decoded_vector.h>
#include <pollux/vector/flat_vector.h>

namespace kumo::pollux::exec {
    template<typename T>
    struct VectorReader {
        using exec_in_t = typename VectorExec::template resolver<T>::in_type;
        // Types without views cannot contain null, they can only be null, so they're
        // in_type is already null_free.
        using exec_null_free_in_t =
        typename VectorExec::template resolver<T>::in_type;

        explicit VectorReader(const DecodedVector *decoded) : decoded_(*decoded) {
        }

        explicit VectorReader(const VectorReader<T> &) = delete;

        VectorReader<T> &operator=(const VectorReader<T> &) = delete;

        vector_size_t index(vector_size_t idx) const {
            return decoded_.index(idx);
        }

        exec_in_t operator[](size_t offset) const {
            return decoded_.template value_at<exec_in_t>(offset);
        }

        exec_null_free_in_t readNullFree(size_t offset) const {
            return decoded_.template value_at<exec_null_free_in_t>(offset);
        }

        bool isSet(size_t offset) const {
            return !decoded_.is_null_at(offset);
        }

        bool may_have_nulls() const {
            return decoded_.may_have_nulls();
        }

        // These functions can be used to check if any elements in a given row are
        // NULL. They are not especially fast, so they should only be used when
        // necessary, and other options, e.g. calling may_have_nulls_recursive() on the
        // vector, have already been exhausted.
        inline bool containsNull(vector_size_t index) const {
            return decoded_.is_null_at(index);
        }

        bool containsNull(vector_size_t startIndex, vector_size_t endIndex) const {
            // Note: This can be optimized for the special case where the underlying
            // vector is flat using bit operations on the nulls buffer.
            for (auto index = startIndex; index < endIndex; ++index) {
                if (containsNull(index)) {
                    return true;
                }
            }

            return false;
        }

        inline bool may_have_nulls_recursive() const {
            return decoded_.may_have_nulls();
        }

        // Scalars don't have children, so this is a no-op.
        void setChildrenMayHaveNulls() {
        }

        const BaseVector *baseVector() const {
            return decoded_.base();
        }

        const DecodedVector &decoded_;
    };

    // ConstantVectorReader and FlatVectorReader are optimized for primitive types
    // in constant or flat encoded vectors.  They operate directly on the vector's
    // content avoiding the need to go through the expensive decoding process.
    template<typename T>
    struct ConstantVectorReader {
        using exec_in_t = typename VectorExec::template resolver<T>::in_type;

        std::optional<exec_in_t> value;

        explicit ConstantVectorReader<T>(ConstantVector<exec_in_t> &vector) {
            if (!vector.is_null_at(0)) {
                value = *vector.rawValues();
            }
        }

        exec_in_t operator[](vector_size_t) const {
            return *value;
        }

        exec_in_t readNullFree(vector_size_t) const {
            return *value;
        }

        bool isSet(vector_size_t) const {
            return value.has_value();
        }

        bool may_have_nulls() const {
            return !value.has_value();
        }

        bool containsNull(vector_size_t) const {
            return !value.has_value();
        }

        bool containsNull(vector_size_t, vector_size_t) const {
            return !value.has_value();
        }

        inline bool may_have_nulls_recursive() const {
            return !value.has_value();
        }

        // Scalars don't have children, so this is a no-op.
        void setChildrenMayHaveNulls() {
        }
    };

    template<typename T>
    struct FlatVectorReader {
        using exec_in_t = typename VectorExec::template resolver<T>::in_type;

        const exec_in_t *values;
        FlatVector<exec_in_t> *vector;

        explicit FlatVectorReader<T>(FlatVector<exec_in_t> &baseVector)
            : values(baseVector.rawValues()), vector(&baseVector) {
        }

        exec_in_t operator[](vector_size_t offset) const {
            return values[offset];
        }

        exec_in_t readNullFree(vector_size_t offset) const {
            return operator[](offset);
        }

        bool isSet(vector_size_t offset) const {
            return !vector->is_null_at(offset);
        }

        bool may_have_nulls() const {
            return vector->may_have_nulls();
        }

        bool containsNull(vector_size_t index) const {
            return !isSet(index);
        }

        bool containsNull(vector_size_t startIndex, vector_size_t endIndex) const {
            for (auto index = startIndex; index < endIndex; ++index) {
                if (containsNull(index)) {
                    return true;
                }
            }
            return false;
        }

        bool may_have_nulls_recursive() const {
            return may_have_nulls();
        }

        // Scalars don't have children, so this is a no-op.
        void setChildrenMayHaveNulls() {
        }
    };

    // This VectorReader is optimized for primitive types in constant or flat
    // encoded vectors.  It operates directly on the vector's content avoiding
    // the need to go through the expensive decoding process.
    template<typename T>
    struct ConstantFlatVectorReader {
        using exec_in_t = typename VectorExec::template resolver<T>::in_type;

        explicit ConstantFlatVectorReader(const FlatVector<exec_in_t> *vector)
            : rawValues_(vector->rawValues()),
              rawNulls_(vector->raw_nulls()),
              indexMultiple_(1) {
        }

        explicit ConstantFlatVectorReader(const ConstantVector<exec_in_t> *vector)
            : rawValues_(vector->rawValues()),
              rawNulls_(vector->is_null_at(0) ? &bits::kNull64 : nullptr),
              indexMultiple_(0) {
        }

        explicit ConstantFlatVectorReader(const VectorReader<T> &) = delete;

        VectorReader<T> &operator=(const VectorReader<T> &) = delete;

        exec_in_t operator[](vector_size_t offset) const {
            return rawValues_[offset * indexMultiple_];
        }

        exec_in_t readNullFree(vector_size_t offset) const {
            return operator[](offset);
        }

        bool isSet(vector_size_t offset) const {
            return !rawNulls_ || !bits::isBitNull(rawNulls_, offset * indexMultiple_);
        }

        bool may_have_nulls() const {
            return rawNulls_;
        }

        inline bool containsNull(vector_size_t index) const {
            return !isSet(index);
        }

        bool containsNull(vector_size_t startIndex, vector_size_t endIndex) const {
            for (auto index = startIndex; index < endIndex; ++index) {
                if (containsNull(index)) {
                    return true;
                }
            }

            return false;
        }

        inline bool may_have_nulls_recursive() const {
            return may_have_nulls();
        }

        // Scalars don't have children, so this is a no-op.
        void setChildrenMayHaveNulls() {
        }

    private:
        const exec_in_t *rawValues_;
        const uint64_t *rawNulls_;
        // Flat Vectors use an identity mapping for indices, Constant Vectors map all
        // indices to 0.  This is the same as multiplying by 1 or 0 respectively.
        // We multiply the index by this value to get that mapping.
        vector_size_t indexMultiple_;
    };

    namespace detail {
        template<typename TOut>
        const TOut &getDecoded(const DecodedVector &decoded) {
            auto base = decoded.base();
            return *base->template as<TOut>();
        }

        inline DecodedVector *decode(DecodedVector &decoder, const BaseVector &vector) {
            decoder.decode(vector);
            return &decoder;
        }
    } // namespace detail

    template<typename K, typename V>
    struct VectorReader<Map<K, V> > {
        using exec_in_t = typename VectorExec::template resolver<Map<K, V> >::in_type;
        using exec_null_free_in_t =
        typename VectorExec::template resolver<Map<K, V> >::null_free_in_type;

        explicit VectorReader(const DecodedVector *decoded)
            : decoded_{*decoded},
              vector_(detail::getDecoded<MapVector>(decoded_)),
              offsets_(vector_.rawOffsets()),
              lengths_(vector_.rawSizes()),
              keyReader_{detail::decode(decodedKeys_, *vector_.mapKeys())},
              valReader_{detail::decode(decodedVals_, *vector_.mapValues())} {
        }

        explicit VectorReader(const VectorReader<Map<K, V> > &) = delete;

        VectorReader<Map<K, V> > &operator=(const VectorReader<Map<K, V> > &) = delete;

        exec_in_t operator[](size_t offset) const {
            auto index = decoded_.index(offset);
            return {&keyReader_, &valReader_, offsets_[index], lengths_[index]};
        }

        exec_null_free_in_t readNullFree(size_t offset) const {
            auto index = decoded_.index(offset);
            return {&keyReader_, &valReader_, offsets_[index], lengths_[index]};
        }

        bool isSet(size_t offset) const {
            return !decoded_.is_null_at(offset);
        }

        bool containsNull(vector_size_t index) const {
            POLLUX_DCHECK(
                keysMayHaveNulls_.has_value() && valuesMayHaveNulls_.has_value(),
                "setChildrenMayHaveNulls() should be called before containsNull()");

            auto decodedIndex = decoded_.index(index);

            return decoded_.is_null_at(index) ||
                   (*keysMayHaveNulls_ &&
                    keyReader_.containsNull(
                        offsets_[decodedIndex],
                        offsets_[decodedIndex] + lengths_[decodedIndex])) ||
                   (*valuesMayHaveNulls_ &&
                    valReader_.containsNull(
                        offsets_[decodedIndex],
                        offsets_[decodedIndex] + lengths_[decodedIndex]));
        }

        bool containsNull(vector_size_t startIndex, vector_size_t endIndex) const {
            for (auto index = startIndex; index < endIndex; ++index) {
                if (containsNull(index)) {
                    return true;
                }
            }

            return false;
        }

        inline bool may_have_nulls_recursive() const {
            POLLUX_DCHECK(
                keysMayHaveNulls_.has_value() && valuesMayHaveNulls_.has_value(),
                "setChildrenMayHaveNulls() should be called before may_have_nulls_recursive()");
            return decoded_.may_have_nulls() || *keysMayHaveNulls_ ||
                   *valuesMayHaveNulls_;
        }

        bool may_have_nulls() const {
            return decoded_.may_have_nulls();
        }

        void setChildrenMayHaveNulls() {
            keyReader_.setChildrenMayHaveNulls();
            valReader_.setChildrenMayHaveNulls();

            keysMayHaveNulls_ = keyReader_.may_have_nulls_recursive();
            valuesMayHaveNulls_ = valReader_.may_have_nulls_recursive();
        }

        const BaseVector *baseVector() const {
            return decoded_.base();
        }

        const DecodedVector &decoded_;
        const MapVector &vector_;
        DecodedVector decodedKeys_;
        DecodedVector decodedVals_;

        const vector_size_t *offsets_;
        const vector_size_t *lengths_;
        VectorReader<K> keyReader_;
        VectorReader<V> valReader_;

        std::optional<bool> keysMayHaveNulls_;
        std::optional<bool> valuesMayHaveNulls_;
    };

    template<typename V>
    struct VectorReader<Array<V> > {
        using exec_in_t = typename VectorExec::template resolver<Array<V> >::in_type;
        using exec_null_free_in_t =
        typename VectorExec::template resolver<Array<V> >::null_free_in_type;
        using exec_in_child_t = typename VectorExec::template resolver<V>::in_type;

        explicit VectorReader(const DecodedVector *decoded)
            : decoded_(*decoded),
              vector_(detail::getDecoded<ArrayVector>(decoded_)),
              offsets_{vector_.rawOffsets()},
              lengths_{vector_.rawSizes()},
              childReader_{detail::decode(arrayValuesDecoder_, *vector_.elements())} {
        }

        bool isSet(size_t offset) const {
            return !decoded_.is_null_at(offset);
        }

        exec_in_t operator[](size_t offset) const {
            auto index = decoded_.index(offset);
            return {&childReader_, offsets_[index], lengths_[index]};
        }

        exec_null_free_in_t readNullFree(size_t offset) const {
            auto index = decoded_.index(offset);
            return {&childReader_, offsets_[index], lengths_[index]};
        }

        inline bool containsNull(vector_size_t index) const {
            POLLUX_DCHECK(
                valuesMayHaveNulls_.has_value(),
                "setChildrenMayHaveNulls() should be called before containsNull()");

            auto decodedIndex = decoded_.index(index);

            return decoded_.is_null_at(index) ||
                   (*valuesMayHaveNulls_ &&
                    childReader_.containsNull(
                        offsets_[decodedIndex],
                        offsets_[decodedIndex] + lengths_[decodedIndex]));
        }

        bool containsNull(vector_size_t startIndex, vector_size_t endIndex) const {
            for (auto index = startIndex; index < endIndex; ++index) {
                if (containsNull(index)) {
                    return true;
                }
            }

            return false;
        }

        inline bool may_have_nulls_recursive() const {
            POLLUX_DCHECK(
                valuesMayHaveNulls_.has_value(),
                "setChildrenMayHaveNulls() should be called before may_have_nulls_recursive()");

            return decoded_.may_have_nulls() || *valuesMayHaveNulls_;
        }

        bool may_have_nulls() const {
            return decoded_.may_have_nulls();
        }

        void setChildrenMayHaveNulls() {
            childReader_.setChildrenMayHaveNulls();

            valuesMayHaveNulls_ = childReader_.may_have_nulls_recursive();
        }

        const BaseVector *baseVector() const {
            return decoded_.base();
        }

        DecodedVector arrayValuesDecoder_;
        const DecodedVector &decoded_;
        const ArrayVector &vector_;
        const vector_size_t *offsets_;
        const vector_size_t *lengths_;
        VectorReader<V> childReader_;
        std::optional<bool> valuesMayHaveNulls_;
    };

    template<typename... T>
    struct VectorReader<Row<T...> > {
        using in_vector_t = RowVector;
        using exec_in_t = typename VectorExec::resolver<Row<T...> >::in_type;
        using exec_null_free_in_t =
        typename VectorExec::template resolver<Row<T...> >::null_free_in_type;

        explicit VectorReader(const DecodedVector *decoded)
            : decoded_(*decoded),
              vector_(detail::getDecoded<in_vector_t>(decoded_)),
              childrenDecoders_{vector_.childrenSize()},
              childReaders_{
                  prepareChildReaders(
                      vector_,
                      std::make_index_sequence<sizeof...(T)>{})
              } {
        }

        exec_in_t operator[](size_t offset) const {
            auto index = decoded_.index(offset);
            return {&childReaders_, index};
        }

        exec_null_free_in_t readNullFree(size_t offset) const {
            auto index = decoded_.index(offset);
            return {&childReaders_, index};
        }

        bool isSet(size_t offset) const {
            return !decoded_.is_null_at(offset);
        }

        bool containsNull(vector_size_t index) const {
            if (decoded_.is_null_at(index)) {
                return true;
            }

            bool fieldsContainNull = false;
            auto decodedIndex = decoded_.index(index);
            std::apply(
                [&](const auto &... reader) {
                    fieldsContainNull |= (reader->containsNull(decodedIndex) || ...);
                },
                childReaders_);

            return fieldsContainNull;
        }

        bool containsNull(vector_size_t startIndex, vector_size_t endIndex) const {
            for (auto index = startIndex; index < endIndex; ++index) {
                if (containsNull(index)) {
                    return true;
                }
            }

            return false;
        }

        inline bool may_have_nulls_recursive() const {
            return decoded_.may_have_nulls_recursive();
        }

        bool may_have_nulls() const {
            return decoded_.may_have_nulls();
        }

        void setChildrenMayHaveNulls() {
            std::apply(
                [](auto &... reader) { (reader->setChildrenMayHaveNulls(), ...); },
                childReaders_);
        }

        const BaseVector *baseVector() const {
            return decoded_.base();
        }

    private:
        template<size_t... I>
        std::tuple<std::unique_ptr<VectorReader<T> >...> prepareChildReaders(
            const in_vector_t &vector,
            std::index_sequence<I...>) {
            return {
                std::make_unique<VectorReader<T> >(
                    detail::decode(childrenDecoders_[I], *vector_.childAt(I)))...
            };
        }

        const DecodedVector &decoded_;
        const in_vector_t &vector_;
        std::vector<DecodedVector> childrenDecoders_;
        std::tuple<std::unique_ptr<VectorReader<T> >...> childReaders_;
    };

    template<typename T>
    struct VectorReader<Variadic<T> > {
        using exec_in_t = typename VectorExec::resolver<Variadic<T> >::in_type;
        using exec_null_free_in_t =
        typename VectorExec::template resolver<Variadic<T> >::null_free_in_type;

        explicit VectorReader(
            std::vector<std::optional<LocalDecodedVector> > &decodedArgs,
            int32_t startPosition)
            : childReaders_{prepareChildReaders(decodedArgs, startPosition)} {
        }

        exec_in_t operator[](vector_size_t offset) const {
            return {&childReaders_, offset};
        }

        exec_null_free_in_t readNullFree(vector_size_t offset) const {
            return {&childReaders_, offset};
        }

        bool isSet(size_t /*unused*/) const {
            // The Variadic itself can never be null, only the values of the underlying
            // Types
            return true;
        }

        bool containsNull(vector_size_t index) const {
            for (const auto &childReader: childReaders_) {
                if (childReader->containsNull(index)) {
                    return true;
                }
            }

            return false;
        }

        bool containsNull(vector_size_t startIndex, vector_size_t endIndex) const {
            for (const auto &childReader: childReaders_) {
                if (childReader->containsNull(startIndex, endIndex)) {
                    return true;
                }
            }

            return false;
        }

        inline bool may_have_nulls_recursive() const {
            for (const auto &childReader: childReaders_) {
                if (childReader->may_have_nulls_recursive()) {
                    return true;
                }
            }

            return false;
        }

        bool may_have_nulls() const {
            for (const auto &childReader: childReaders_) {
                if (childReader->may_have_nulls()) {
                    return true;
                }
            }
            return false;
        }

        void setChildrenMayHaveNulls() {
            for (auto &childReader: childReaders_) {
                childReader->setChildrenMayHaveNulls();
            }
        }

    private:
        auto prepareChildReaders(
            std::vector<std::optional<LocalDecodedVector> > &decodedArgs,
            int32_t startPosition) {
            std::vector<std::unique_ptr<VectorReader<T> > > childReaders;
            childReaders.reserve(decodedArgs.size() - startPosition);

            for (int i = startPosition; i < decodedArgs.size(); ++i) {
                childReaders.emplace_back(
                    std::make_unique<VectorReader<T> >(decodedArgs.at(i).value().get()));
            }

            return childReaders;
        }

        std::vector<std::unique_ptr<VectorReader<T> > > childReaders_;
    };

    template<typename T, bool comparable, bool orderable>
    struct VectorReader<Generic<T, comparable, orderable> > {
        using exec_in_t = GenericView;
        using exec_null_free_in_t = exec_in_t;

        template<TypeKind kind>
        void ensureCastedReader() {
            if constexpr (TypeTraits<kind>::isPrimitiveType) {
                this->operator[](0)
                        .template ensureReader<typename KindToSimpleType<kind>::type>();
            }
        }

        explicit VectorReader(const DecodedVector *decoded) : decoded_(*decoded) {
            if (decoded->size() && decoded->base()->type()->isPrimitiveType()) {
                TypeKind kind = decoded->base()->type_kind();
                POLLUX_DYNAMIC_TYPE_DISPATCH_ALL(ensureCastedReader, kind);
            }
        }

        explicit VectorReader(
            const VectorReader<Generic<T, comparable, orderable> > &) = delete;

        VectorReader<Generic<T, comparable, orderable> > &operator=(
            const VectorReader<Generic<T, comparable, orderable> > &) = delete;

        bool isSet(vector_size_t offset) const {
            return !decoded_.is_null_at(offset);
        }

        exec_in_t operator[](vector_size_t offset) const {
            // We pass the non-decoded index.
            return GenericView{decoded_, castReaders_, castType_, offset};
        }

        exec_null_free_in_t readNullFree(vector_size_t offset) const {
            return operator[](offset);
        }

        inline bool containsNull(vector_size_t /* index */) const {
            // This function is only called if callNullFree is defined.
            // TODO (kevinwilfong): Add support for Generics in callNullFree.
            POLLUX_UNSUPPORTED(
                "Calling callNullFree with Generic arguments is not yet supported.");
        }

        bool containsNull(
            vector_size_t /* startIndex */,
            vector_size_t /* endIndex */) const {
            // This function is only called if callNullFree is defined.
            // TODO (kevinwilfong): Add support for Generics in callNullFree.
            POLLUX_UNSUPPORTED(
                "Calling callNullFree with Generic arguments is not yet supported.");
        }

        inline bool may_have_nulls_recursive() const {
            // This function is only called if callNullFree is defined.
            // TODO (kevinwilfong): Add support for Generics in callNullFree.
            POLLUX_UNSUPPORTED(
                "Calling callNullFree with Generic arguments is not yet supported.");
        }

        bool may_have_nulls() const {
            return decoded_.may_have_nulls();
        }

        inline void setChildrenMayHaveNulls() {
            // This function is only called if callNullFree is defined.
            // TODO (kevinwilfong): Add support for Generics in callNullFree.
            POLLUX_UNSUPPORTED(
                "Calling callNullFree with Generic arguments is not yet supported.");
        }

        const BaseVector *baseVector() const {
            return decoded_.base();
        }

        const DecodedVector &decoded_;

        // Those two variables are mutated by the GenericView during cast operations,
        // and are shared across GenericViews constructed by the reader.
        mutable std::array<std::shared_ptr<void>, 3> castReaders_;
        mutable std::optional<const std::type_info *> castType_ = std::nullopt;
    };

    template<>
    struct VectorReader<DynamicRow> {
        using in_vector_t = RowVector;
        using exec_in_t = DynamicRowView<true>;
        using exec_null_free_in_t = DynamicRowView<false>;

        explicit VectorReader(const DecodedVector *decoded)
            : decoded_(*decoded),
              vector_(detail::getDecoded<in_vector_t>(decoded_)),
              childrenDecoders_{vector_.childrenSize()} {
            for (int i = 0; i < vector_.childrenSize(); i++) {
                childReaders_.push_back(std::make_unique<VectorReader<Any> >(
                    detail::decode(childrenDecoders_[i], *vector_.childAt(i))));
            }
        }

        exec_in_t operator[](size_t offset) const {
            auto index = decoded_.index(offset);
            return {&childReaders_, index};
        }

        exec_null_free_in_t readNullFree(size_t offset) const {
            auto index = decoded_.index(offset);
            return {&childReaders_, index};
        }

        bool isSet(size_t offset) const {
            return !decoded_.is_null_at(offset);
        }

        bool may_have_nulls() const {
            return decoded_.may_have_nulls();
        }

        const BaseVector *baseVector() const {
            return decoded_.base();
        }

    private:
        const DecodedVector &decoded_;
        const in_vector_t &vector_;
        std::vector<DecodedVector> childrenDecoders_;
        std::vector<std::unique_ptr<VectorReader<Any> > > childReaders_;
    };

    template<typename T, bool providesCustomComparison>
    struct VectorReader<CustomType<T, providesCustomComparison> >
            : public VectorReader<typename T::type> {
        explicit VectorReader(const DecodedVector *decoded)
            : VectorReader<typename T::type>(decoded) {
        }
    };

    template<typename T>
    struct VectorReader<CustomType<T, true> > {
        using exec_underlying_in_t =
        typename VectorExec::template resolver<typename T::type>::in_type;
        using exec_in_t = CustomTypeWithCustomComparisonView<exec_underlying_in_t>;
        // Only custom types around primitive types can provide custom comparison
        // operators, so they cannot contain null, they can only be null, so they're
        // in_type is already null_free.
        using exec_null_free_in_t = exec_in_t;

        explicit VectorReader(const DecodedVector *decoded)
            : decoded_(*decoded),
              customComparisonType_(
                  std::dynamic_pointer_cast<const CanProvideCustomComparisonType<
                      SimpleTypeTrait<exec_underlying_in_t>::typeKind>>(
                      decoded_.base()->type())) {
        }

        explicit VectorReader(const VectorReader<CustomType<T, true> > &) = delete;

        CustomType<T, true> &operator=(const CustomType<T, true> &) = delete;

        exec_in_t operator[](size_t offset) const {
            return exec_in_t(
                decoded_.template value_at<exec_underlying_in_t>(offset),
                customComparisonType_);
        }

        exec_null_free_in_t readNullFree(size_t offset) const {
            return exec_null_free_in_t(
                decoded_.template value_at<exec_underlying_in_t>(offset),
                customComparisonType_);
        }

        bool isSet(size_t offset) const {
            return !decoded_.is_null_at(offset);
        }

        bool may_have_nulls() const {
            return decoded_.may_have_nulls();
        }

        // These functions can be used to check if any elements in a given row are
        // NULL. They are not especially fast, so they should only be used when
        // necessary, and other options, e.g. calling may_have_nulls_recursive() on the
        // vector, have already been exhausted.
        inline bool containsNull(vector_size_t index) const {
            return decoded_.is_null_at(index);
        }

        bool containsNull(vector_size_t startIndex, vector_size_t endIndex) const {
            // Note: This can be optimized for the special case where the underlying
            // vector is flat using bit operations on the nulls buffer.
            for (auto index = startIndex; index < endIndex; ++index) {
                if (containsNull(index)) {
                    return true;
                }
            }

            return false;
        }

        inline bool may_have_nulls_recursive() const {
            return decoded_.may_have_nulls();
        }

        // Scalars don't have children, so this is a no-op.
        void setChildrenMayHaveNulls() {
        }

        const BaseVector *baseVector() const {
            return decoded_.base();
        }

        const DecodedVector &decoded_;
        const std::shared_ptr<const CanProvideCustomComparisonType<
            SimpleTypeTrait<exec_underlying_in_t>::typeKind>>
        customComparisonType_;
    };
} // namespace kumo::pollux::exec
